import datetime
import os
import sys
sys.path.append(os.getcwd())
from rocketmq.client import Producer, Message
from rocketmq.client import PushConsumer, PullConsumer
import pandas as pd
import time
import json
import numpy as np
# import threading
from src.utils.config import config
from src.utils.config import logger
from datetime import datetime
import uuid
# from flask import Flask
# from flask_restful import Api, Resource
# app = Flask(__name__)
# api = Api(app)


class PriceIndexActMQ():

    def __init__(self):
        self.addr = config.get_config("dqc_montior_mq", "addr")
        self.group_id = config.get_config("dqc_montior_mq", "premiumRateGroup")
        self.topic = config.get_config("dqc_montior_mq", "commonTopic")
        self.send_tag = config.get_config("dqc_montior_mq", "premiumRateReceiveTag")
        self.receive_tag = config.get_config("dqc_montior_mq", "premiumRateSendTag")
        self.accessKey = config.get_config("dqc_montior_mq", "accessKey")
        self.secretKey = config.get_config("dqc_montior_mq", "secretKey")

        self.global_producer = Producer(self.group_id)
        self.global_producer.set_namesrv_addr(self.addr)
        if self.accessKey is not None and self.secretKey is not None:
            self.global_producer.set_session_credentials(self.accessKey, self.secretKey, 'ALIYUN1')
        self.global_producer.start()

        self.price_index = {}
        self.price_waiting = False

    def producer(self, val):

        msg = Message(self.topic)
        msg.set_tags(self.receive_tag)
        msg.set_body(val)
        msg.set_keys('test')
        ret = self.global_producer.send_sync(msg)
        print(ret.status, ret.msg_id, ret.offset)
        self.global_producer.shutdown()

    '''
    消费方式PullConsumer（全部消费）（可重复消费）
    '''

    def pull_consumer(self):
        consumer = PullConsumer(self.group_id)
        consumer.set_namesrv_addr(self.addr)
        if self.accessKey is not None and self.secretKey is not None:
            consumer.set_session_credentials(self.accessKey, self.secretKey, 'ALIYUN')
        consumer.start()
        for msg in consumer.pull(self.topic, self.receive_tag):
            self.process(msg)
        consumer.shutdown()

    '''
    消费方式PushConsumer（即时消费）（不可重复消费）
    '''

    def push_consumer(self):
        consumer = PushConsumer(self.group_id)
        consumer.set_namesrv_addr(self.addr)
        if self.accessKey is not None and self.secretKey is not None:
            consumer.set_session_credentials(self.accessKey, self.secretKey, 'ALIYUN')
        consumer.subscribe(self.topic, self.process, self.receive_tag)
        consumer.start()
        while True:
            time.sleep(1* 60)
            # self.refresh_price_index()


    def say_hello(self, msg):
        ts = msg.store_timestamp / 1000
        logger.info("push_consumer store_time:{} key:{} tags:{} body:{}".format(
            datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S"),
            msg.keys, msg.tags, msg.body))

    def process(self, msg):
        # while self.price_waiting:
        #     logger.info("push_consumer process waiting {}".format(datetime.datetime.now()))
        #     time.sleep(20)
        logger.info('recive keys {} body {}'.format(msg.keys,msg))
        notice = json.loads(msg.body)
        ts = msg.store_timestamp / 1000
        if notice.get('tag') is None:
            logger.info(
                "store_time {} pull_consumer loss receive_tag key:{} tags:{} body:{}".format(
                    datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S"),
                    msg.keys, msg.tags,
                    msg.body))
            return
        data, warn_data, warn_msg = [], [], []
        if notice.get('type') == "product" :
            if notice.get('productIdList') is not None:
                data, warn_data, warn_msg = self.cal_product(notice.get('productIdList'))
        elif notice.get('type') == "brand"  :
            if notice.get('brandIdList') is not None:
                data, warn_data, warn_msg = self.cal_brand(notice.get('brandIdList'))

        # if notice.get('type') == "category" and notice.get('categoryIdList') is not None:
        #     data, warn_data, warn_msg = self.cal_category(notice.get('categoryIdList'))

        logger.warn('\nstore_time {} key:{} warn_size:{} warn_msg:{} warn_data:{}\n'.format(
            datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S"),
            msg.keys, len(warn_msg), str(warn_msg), str(warn_data)))

        self.index_producer(notice.get('type'), notice.get('tag'),
                            {"batchNo": notice.get("batchNo"),
                             "type": notice.get('type'),
                             "data": data},msg)

    def index_producer(self, key, tag, val,msg):
        try:
            body = json.dumps(val)
            msg = Message(self.topic)
            msg.set_keys(key)
            msg.set_tags(tag)
            msg.set_body(body)
            self.global_producer.send_sync(msg)
        # ret = self.global_producer.send_sync(msg)
        # logger.info("status {} msg_id {} offset {}".format(ret.status, ret.msg_id, ret.offset))
        except Exception as e:
            logger.critical('mq 数据报错：@{},{}'.format(e,msg.keys))
            raise TypeError('等级比率报错:') from e

    def dqc_producer_msg(self, ruleCode,serverId):
        try:
            unique_id = uuid.uuid4()
            mq_val = {
                "serverId": serverId,
                "serverMonitorRuleCode": ruleCode,
                "serverRecordId": unique_id.hex,
                "isStart": 1,
                "probeTime": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            }
            body = json.dumps(mq_val)
            msg = Message(self.topic)
            msg.set_keys(self.send_tag)
            msg.set_tags(self.send_tag)
            msg.set_body(body)
            # self.global_producer.send_sync(msg)
            self.global_producer.send_sync(msg)
        except Exception as e:
            logger.critical('mq 数据报错：@{},{}'.format(e,msg.keys))
            raise TypeError('等级比率报错:') from e

    # status 0:成功 -1:错误
    def cal_product(self, productIdList):
        data, warn_data, warn_msg = [], [], []
        for item in productIdList:
            productId = item.get("productId")
            ukc = item.get("ukc")
            ratio = 0.0
            try:
                productId = int(productId)
                premium_df = pd.Dataframe()
                match_data = premium_df.loc[premium_df.product_id==productId]
                if match_data.empty:
                    #test
                    # ratio=productId%5*0.01
                    ratio=0
                else:
                    match_data.reset_index(inplace=True)
                    ratio =match_data.loc[0,'premium_rate']
                    # ratio = random.uniform(-0.1, 0.1)
                # tmp_old = self.price_index[day][self.price_index[day]['product_id'] == productId]
                # if tmp_old.shape[0] != 1:
                #     raise Exception("not find productId")
                # old = tmp_old['price_index'].iloc[0]
                # if np.isnan(old):
                #     raise Exception("not find")

                # tmp_new = self.price_index[1][self.price_index[1]['product_id'] == productId]
                # if tmp_new.shape[0] != 1:
                #     raise Exception("not find productId")
                # new = tmp_new['price_index'].iloc[0]
                # if np.isnan(new):
                #     raise Exception("not find")

                # ratio = (new - old) / ((day - 1) * old)
                # if np.isnan(ratio):
                #     raise Exception("not find")
                # if new * old == 0:
                #     raise Exception("price index is zero")
                data.append({"productId": productId,
                             "premiumRate": "{}".format(round(ratio, 4)),
                             "ukc": ukc,
                             "status": "0"})
                logger.info('item {} ratio {}'.format(str(item), round(ratio, 4)))
            except Exception as e:
                logger.warn('item {} msg {}'.format(str(item), str(e)))
                data.append({"productId": productId,
                             "premiumRate": "{}".format(round(ratio, 4)),
                             "ukc": ukc,
                             "status": "-1",
                             "msg": str(e)})
                warn_data.append(productId)
                warn_msg.append(str(e))
        return data, warn_data, warn_msg

    # status 0:成功 -1:错误
    def cal_brand(self, brandIdList):
        data, warn_data, warn_msg = [], [], []
        for item in brandIdList:
            brandId = item.get("brandId")
            categoryId = item.get("categoryId")
            exculdeProductIdList = item.get("exculdeProductIdList", [])
            ukc = item.get("ukc")
            priceRange = item.get("priceRange")
            ratio=0.0
            try:
                brandId=int(brandId)
                categoryId=int(categoryId)
                # product_id, paipai_settle_amount_sum, bi_sale_price_sum, settle_amount_num_sum,
                # pp_cnt, pjt_cnt, premium_rate, product_category_id, price_segment, product_brand_id, insert_date
                premium_df = PRODUCT_PREMIUM_RATE_DF.copy()
                match_data = premium_df.loc[(premium_df.price_segment==priceRange) & (premium_df.product_brand_id==brandId) & (premium_df.product_category_id==categoryId) & (~premium_df['product_id'].isin(list(exculdeProductIdList)))]
                if match_data.empty:
                    #test
                    # ratio=categoryId%5*0.01
                    ratio=0
                    # pass
                else:
                    paipai_settle_amount_sum = match_data.paipai_settle_amount_sum.sum()
                    bi_sale_price_sum = match_data.bi_sale_price_sum.sum()
                    settle_amount_num_sum = match_data.settle_amount_num_sum.sum()
                    ratio = (paipai_settle_amount_sum+settle_amount_num_sum)*1.0/(bi_sale_price_sum+settle_amount_num_sum)-1
                    # ratio = random.uniform(-0.1, 0.1)

                # data.append({"brandId": brandId,
                #              "categoryId": categoryId,
                #              "premiumRate": "{}".format(round(ratio, 4)),
                #              "ukc": ukc,
                #              "status": "0"})
                item['status']='0'
                item['premiumRate']=str(round(ratio, 4))
                data.append(item)
                logger.info('item {} ratio {}'.format(str(item), round(ratio, 4)))
            except Exception as e:
                logger.warn('item {} msg {}'.format(str(item), str(e)))
                # data.append({"brandId": brandId,
                #              "categoryId": categoryId,
                #              "premiumRate": "{}".format(round(ratio, 4)),
                #              "ukc": ukc,
                #              "status": "-1",
                #              "msg": str(e)})
                item['status'] = '1'
                item['premiumRate'] = str(round(ratio, 4))
                data.append(item)
                warn_data.append((brandId, categoryId))
                warn_msg.append(str(e))
        return data, warn_data, warn_msg

    # status 0:成功 -1:错误
    def cal_category(self, categoryIdList):
        data, warn_data, warn_msg = [], [], []
        for item in categoryIdList:
            categoryId = item.get("categoryId")
            ukc = item.get("ukc")
            try:
                day = int(item.get("day"))
                exculdeBrandIdList = [str(brand_id) for brand_id in item.get("exculdeBrandIdList", [])]
                exculdeProductIdList = [str(product_id) for product_id in item.get("exculdeProductIdList", [])]

                tmp_old = self.price_index[day][self.price_index[day]['product_category_id'] == categoryId]
                tmp_old = tmp_old[(~tmp_old['product_brand_id'].isin(exculdeBrandIdList)) &
                                  (~tmp_old['product_id'].isin(exculdeProductIdList))]
                if tmp_old.shape[0] == 0:
                    raise Exception("not find categoryId")

                old = (tmp_old['weight'] * tmp_old['price_index']).sum() / tmp_old['weight'].sum()
                if np.isnan(old):
                    raise Exception("not find")

                tmp_new = self.price_index[1][self.price_index[1]['product_category_id'] == categoryId]
                tmp_new = tmp_new[(~tmp_new['product_brand_id'].isin(exculdeBrandIdList)) &
                                  (~tmp_new['product_id'].isin(exculdeProductIdList))]
                if tmp_new.shape[0] == 0:
                    raise Exception("not find categoryId")

                new = (tmp_new['weight'] * tmp_new['price_index']).sum() / tmp_new['weight'].sum()
                if np.isnan(new):
                    raise Exception("not find")

                ratio = (new - old) / ((day - 1) * old)
                if np.isnan(ratio):
                    raise Exception("not find")
                if new * old == 0:
                    raise Exception("price index is zero")
                data.append({"categoryId": categoryId,
                             "ratio": "{}".format(round(ratio, 4)),
                             "ukc": ukc,
                             "status": "0"})
                logger.info('item {} ratio {}'.format(str(item), round(ratio, 4)))
            except Exception as e:
                logger.warn('item {} msg {}'.format(str(item), str(e)))
                data.append({"categoryId": categoryId,
                             "ratio": "{}".format(""),
                             "ukc": ukc,
                             "status": "-1",
                             "msg": str(e)})
                warn_data.append(categoryId)
                warn_msg.append(str(e))
        return data, warn_data, warn_msg

# def start_ActMQ():
#     model = PriceIndexActMQ()
#     # model.pull_consumer()
#     model.push_consumer()

product_mq_model = PriceIndexActMQ()
# product_mq_model.dqc_producer_msg('tablet_model_trainer',
#                                   'tablet_model_trainer')
#运行下面的函数必须把上面的321行注释掉
# if __name__=='__main__':
#     model = PriceIndexActMQ()
#     # model.pull_consumer()
#
#     model.dqc_producer_msg('sku_level_mobile_train','test1')


